package com.shxzz.easy12306.services.pay.mq.base;

import cn.hutool.core.text.CharSequenceUtil;
import com.alibaba.fastjson2.JSON;
import com.shxzz.easy12306.framework.starter.common.mq.base.BaseSendExtendDTO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;

import java.util.Optional;

/**
 * RocketMQ 通用消息发送模板类
 * @author shxzz
 * @create 2024-03-09-23:56
 */
@Slf4j
@RequiredArgsConstructor
public abstract class AbstractCommonProduceSendTemplate<T> {

    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 构建消息发送基本扩展信息
     */
    protected abstract BaseSendExtendDTO buildBaseSendExtendDTO(T messageSendEvent);

    /**
     * 构建消息体
     */
    protected abstract Message<?> buildMessage(T messageSendEvent, BaseSendExtendDTO baseSendExtendDTO);

    public SendResult sendMessage(T messageSendEvent) {
        BaseSendExtendDTO baseSendExtendDTO = buildBaseSendExtendDTO(messageSendEvent);
        String destination = baseSendExtendDTO.getTopic();
        if(CharSequenceUtil.isNotEmpty(baseSendExtendDTO.getTag())) {
            destination += ":" + baseSendExtendDTO.getTag();
        }
        try {
            SendResult sendResult = rocketMQTemplate.syncSend(
                    destination,
                    buildMessage(messageSendEvent, baseSendExtendDTO),
                    baseSendExtendDTO.getSendTimeout(),
                    Optional.ofNullable(baseSendExtendDTO.getDelayLevel()).orElse(0));
            log.info("[{}] 消息发送结果：{}，消息ID：{}，消息Keys：{}", baseSendExtendDTO.getEventName(), sendResult.getSendStatus(), sendResult.getMsgId(), baseSendExtendDTO.getKeys());
            return sendResult;
        } catch (Exception e) {
            log.error("[{}] 消息发送失败，消息体：{}", baseSendExtendDTO.getEventName(), JSON.toJSONString(messageSendEvent), e);
            throw e;
        }
    }
}
